package com.mchat.user.dao;

import com.huaban.analysis.jieba.SegToken;
import com.mchat.user.domain.dto.UserInfoDTO;
import com.mchat.user.domain.dto.UserPrivacyDTO;
import com.mchat.user.domain.po.User;
import io.r2dbc.spi.Result;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import io.r2dbc.spi.Statement;
import lombok.RequiredArgsConstructor;
import org.reactivestreams.Publisher;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.function.BiFunction;

@RequiredArgsConstructor
@Component
public class UserDAO {
    private final DatabaseClient databaseClient;
    private final BiFunction<Row, RowMetadata, User> MAP_FUNCTION = (row, rowMetadata) -> {
        User user = new User();
        user.setId(row.get("id", Long.class));
        user.setName(row.get("name", String.class));
        user.setPassword(row.get("password", String.class));
        user.setEmail(row.get("email", String.class));
        user.setProfileImg(row.get("profile_img", String.class));
        user.setStatus(row.get("status", Integer.class));
        user.setCreateTime(row.get("create_time", LocalDateTime.class));
        user.setGender(row.get("gender", Integer.class));
        user.setFromRegion(row.get("from_region", String.class));
        user.setFromSubregion(row.get("from_subregion", String.class));
        user.setPresentRegion(row.get("present_region", String.class));
        user.setPresentSubregion(row.get("present_subregion", String.class));
        user.setBirthday(row.get("birth_day", LocalDate.class));
        user.setLastUpdate(row.get("last_update", LocalDateTime.class));
        user.setProfileVisible(row.get("is_profile_visible", Boolean.class));
        user.setSearchableByEmail(row.get("is_searchable_by_email", Boolean.class));
        return user;
    };
    //需要查询的字段
    String tableField= """
            id, name, password, email, profile_img, status, 
            create_time, gender, birth_day, from_region, 
            present_region, from_subregion, present_subregion,
            last_update, is_profile_visible, is_searchable_by_email
            """;

    public Mono<User> selectByEmail(String email) {

        return databaseClient.sql("select "+tableField+" from tb_user where email= $1")
                .bind(0, email)
                .map(MAP_FUNCTION)
                .one();
    }

    public Mono<User> getById(Long id) {
        return databaseClient.sql("select "+tableField+" from tb_user where id= $1")
                .bind(0, id)
                .map(MAP_FUNCTION)
                .one();
    }

    public Mono<Boolean> SelectExist(String email) {
        return databaseClient.sql("SELECT EXISTS(SELECT 1 FROM tb_user WHERE email = $1)")
                .bind(0, email)
                .map(row -> row.get(0, Boolean.class))
                .one();
    }


    public Flux<User> getBatchUser(Flux<Long> ids) {
        String sql = "SELECT "+ tableField+" FROM tb_user WHERE id = ANY($1)";
        return ids
                .buffer(50) // 每 50 个 id 批处理一次
                .flatMap(list -> {
                            if (list.size() == 0) return Flux.empty();
                            return databaseClient.inConnectionMany(connection -> {
                                Statement statement = connection.createStatement(sql);
                                // 把 list 转成 Long[]，绑定到 $1
                                statement.bind(0, list.toArray(new Long[0]));
                                // fetchSize 设置一次读取的数量
                                return Flux.from(statement.fetchSize(50).execute())
                                        .flatMap(result -> result.map(MAP_FUNCTION));
                            });
                        }
                );
    }


    //修改可更改部分的用户信息
    public Mono<Long> updateInfo(long id, UserInfoDTO dto) {
        String sql = """
                update tb_user set  birth_day = $1,from_region=$2,from_subregion=$3,present_region=$4,present_subregion=$5 ,gender=$6 where id=$7
                """;
        return databaseClient.inConnection(connection -> {
            Statement statement = connection.createStatement(sql);
            statement.bind(1, dto.getFromRegion())
                    .bind(2, dto.getFromSubregion())
                    .bind(3, dto.getPresentRegion())
                    .bind(4, dto.getPresentSubregion())
                    .bind(5, dto.getGender())
                    .bind(6, id);

            if (dto.getBirthday() == null) {
                statement.bindNull(0, LocalDateTime.class);
            } else {
                statement.bind(0, dto.getBirthday());
            }
            var execute = statement.execute();
            return Mono.from(execute)
                    .flatMap(result -> Mono.from(result.getRowsUpdated()));
        });
    }

    public Mono<Long> updatePwd(String email, String pwd) {
        return databaseClient.inConnection(connection -> {
            Statement statement = connection.createStatement("update tb_user set password=$1,last_update=current_timestamp where email=$2");
            return Mono.from(statement.bind(0, pwd)
                            .bind(1, email)
                            .execute())
                    .flatMap(result -> Mono.from(result.getRowsUpdated()));
        });
    }

    public Mono<Void> updateAvatar(long id, String avatar) {
        return databaseClient.inConnection(connection -> {
            var execute = connection.createStatement("update tb_user set profile_img=$1 where id=$2")
                    .bind(0, avatar)
                    .bind(1, id)
                    .execute();
            return Mono.from(execute).flatMap(result -> Mono.from(result.getRowsUpdated())).then();
        });
    }

    public Mono<Long> save(User user) {
        String sql = """
                INSERT INTO tb_user (id,name, password, email, profile_img, status,  create_time, gender ,last_update)
                VALUES ($1,$2,$3, $4, $5, $6, current_timestamp, $7,current_timestamp)
                """;
        return databaseClient.inConnection(connection -> {
            Statement statement = connection.createStatement(sql);
            return Mono.from(
                            statement
                                    .bind(0, user.getId())
                                    .bind(1, user.getName())
                                    .bind(2, user.getPassword())
                                    .bind(3, user.getEmail())
                                    .bind(4, user.getProfileImg())
                                    .bind(5, user.getStatus())
                                    .bind(6, user.getGender())
                                    .execute()
                    )
                    .flatMap(result -> Mono.from(result.getRowsUpdated()));
        });
    }

    public Mono<Long> updatePrivacy(long id, UserPrivacyDTO dto) {
        String sql = """
                update tb_user set  is_profile_visible=$1 , is_searchable_by_email=$2 where id=$3
                """;
        return databaseClient.inConnection(connection -> Mono.from(
                connection.createStatement(sql)
                        .bind(0, dto.getProfileVisible())
                        .bind(1, dto.getSearchableByEmail())
                        .bind(2, id)
                        .execute()
        ).flatMap(result -> Mono.from(result.getRowsUpdated())));
    }

    public Mono<Long> updateName(long id, String name, List<SegToken> tokens) {
        String sql = """
                UPDATE tb_user
                SET name = $1,name_tsv = to_tsvector('simple',$2)
                WHERE id = $3;
                """;
        return databaseClient.inConnection(connection -> {
            Publisher<? extends Result> execute;
            if (tokens != null) {
                StringBuilder stringBuilder = new StringBuilder(name.length() * 3);
                for (SegToken token : tokens) {
                    stringBuilder.append(token.word).append(' ');
                }
                stringBuilder.append(name);

                execute = connection.createStatement(sql)
                        .bind(0, name)
                        .bind(1, stringBuilder.toString())
                        .bind(2, id)
                        .execute();
            } else {
                execute = connection.createStatement(sql)
                        .bind(0, name)
                        .bind(1, name)
                        .bind(2,id)
                        .execute();
            }
            return Mono.from(execute).flatMap(result -> Mono.from(result.getRowsUpdated()));
        });
    }
    public Flux<Long> searchName(String name){
        String sql= """
                    SELECT id
                    FROM tb_user
                    WHERE name_tsv @@   to_tsquery($1) limit 100
                    """;
        return databaseClient.sql(sql)
                .bind(0,name)
                .map(readable -> readable.get(0,Long.class))

                .all();
    }



}
